package cn.mjnxin.iot.connector;

import java.io.File;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;

import cn.mjnxin.iot.base.cache.ProtocalStrategyCache;
import cn.mjnxin.iot.base.domain.NodeConfig;
import cn.mjnxin.iot.base.domain.ProtocolConfig;
import cn.mjnxin.iot.base.log.LogWorker;
import cn.mjnxin.iot.base.util.ConfigTool;
import cn.mjnxin.iot.connector.base.ChannelDataSender;
import cn.mjnxin.iot.connector.base.IChannelDataSender;
import cn.mjnxin.iot.connector.cache.HostRuntimeCache;
import cn.mjnxin.iot.connector.connector.ConnectorAsClient2Gate;
import cn.mjnxin.iot.connector.connector.ConnectorAsServer4Gate;
import cn.mjnxin.iot.connector.context.ConnectorContext;
import cn.mjnxin.iot.connector.context.IPeerChangedListener;
import cn.mjnxin.iot.connector.context.IPeerContext;
import cn.mjnxin.iot.connector.handle.ChannelDataHandle;
import cn.mjnxin.iot.connector.handle.IChannelDataHandle;
import cn.mjnxin.iot.connector.handle.IHandleFactory;
import cn.mjnxin.iot.connector.message.IMessageHandle;
import cn.mjnxin.iot.connector.worker.Connector2GateWorker;
import cn.mjnxin.iot.connector.worker.Gate2ConnectorWorker;
import cn.mjnxin.iot.transfer.ConnectionWacthDog;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;

/**
 * 主连接器
 * @author CLG 2023年10月10日
 * @version V1.0.0
 */
public class Connector implements IConnector {
	private static Map<Integer, ConnectorAsClient2Gate> ConnectorAsClient2Gate = new HashMap<>();
	// <IP:Port,port:pid>
	public  static Map<String, NodeConfig>			    gateAddress	     	   = new LinkedHashMap<>(1);
	private IHandleFactory         messageHandleFactory   = null;
	private ConnectorAsServer4Gate connectorAsServer4Gate = null;
	private IChannelDataHandle	   channelDataHandle	  = null;
	private String				   connectorServerAddress = null;
	private Integer				   connectorServerPort	  = null;
	private String				   connectorApiAddress	  = null;
	private Integer				   connectorServerPid	  = null;  // 根据协议编号选取相应的编解码业务处理服务
	private Integer				   timeout	         	  = null;
	// 日志等级 0：无日志 1：报文长度日志 2：报文内容日志 3：所有日志
	private Integer                logLevel               = 0;
	private boolean	               mngMode				  = false; //管理模式
	private boolean	               devMode				  = false; //开发模式
	private boolean				   isAllMatch			  = false; //全匹配模式
	private IChannelDataSender	   channelDataSender	  = null;
	private String				   connectMode			  = "";
	private boolean				   connectAsClient		  = false;
	private Integer				   workerThreads		  = 10;
	private ConnectorContext	   context				  = new ConnectorContext();
	private static Connector	   me				      = new Connector();
	
	public static IConnector me() {
		return me;
	}
	
	private Connector() {
	}

	public void init(String[] args, IHandleFactory messageHandleFactory, String connectorApiAddress) {
		parseConfig(args.length > 0 ? args[0] : null);

		this.messageHandleFactory = messageHandleFactory;
		this.connectorApiAddress = connectorApiAddress;
		this.channelDataHandle = new ChannelDataHandle(context, messageHandleFactory, "host".equals(connectMode));
		//初始化数据中转线程
		try {
			Connector2GateWorker.init(workerThreads, this.getGateAddress(), "host".equals(connectMode));

		} catch (Exception e) {
			LogWorker.error("Connector2GateWorker.init(1)失败", e);
			System.exit(-1);
		}
		try {
			Gate2ConnectorWorker.init(workerThreads, channelDataHandle);

		} catch (Exception e) {
			LogWorker.error("Connector2GateWorker.init(1, channelDataHandle)失败", e);
			System.exit(-1);
		}
		channelDataSender = new ChannelDataSender(context, this.getGateAddress());
	}

	public void setContextChangedListener(IPeerChangedListener contextChangedListener) {
		this.context.setContextChangedListener(contextChangedListener);
	}
	
	@Override
	public IMessageHandle<?> getMessageHandle(int pid) {
		return messageHandleFactory.getMessageHandle(pid);
	}
	
	@Override
	public IChannelDataSender getChannelDataSender() {
		return channelDataSender;
	}

	@Override
	public IPeerContext getContext() {
		return context;
	}

	@Override
	public Integer getPid() {
		return this.connectorServerPid;
	}

	@Override
	public Integer getLogLevel() {
		return this.logLevel;
	}

	@Override
	public String getGateAddress() {
		if (gateAddress.isEmpty()) {
			return null;
		}
		else {
			 NodeConfig nc = gateAddress.values().iterator().next();
			 
			 return nc.getAddress() + ":" + nc.getPort();
		}
	}

	@Override
	public boolean isDevMode() {
		return this.devMode;
	}
	
	@Override
	public void launch() throws InterruptedException {
		if (this.isAllMatch) {
			Integer[] pidArray = ProtocalStrategyCache.protocalStrategyCache.keySet().toArray(new Integer[ProtocalStrategyCache.protocalStrategyCache.size()]);
			if (pidArray.length == 0) {
				if (gateAddress.isEmpty()) {
					messageHandleFactory.initAllMessageHandle(new Integer[]{this.connectorServerPid});
				}
				else {
					List<Integer> pidList = gateAddress.values().stream().map(NodeConfig::getPid).collect(Collectors.toList());
					messageHandleFactory.initAllMessageHandle(pidList.toArray(new Integer[pidList.size()]));
				}
			}
			else {
				if (gateAddress.isEmpty()) {
					messageHandleFactory.initAllMessageHandle(pidArray);
				}
				else {
					messageHandleFactory.initAllMessageHandle(pidArray);
				}
			}
		}
		else {
			if (gateAddress.isEmpty()) {
				messageHandleFactory.initAllMessageHandle(new Integer[]{this.connectorServerPid});
			}
			else {
				List<Integer> pidList = gateAddress.values().stream().map(NodeConfig::getPid).collect(Collectors.toList());
				messageHandleFactory.initAllMessageHandle(pidList.toArray(new Integer[pidList.size()]));
			}
		}
		messageHandleFactory.startMQListener();
		
		if (connectAsClient) {
			startAsClient(context, mngMode, isAllMatch);
		}
		else {
			connectorAsServer4Gate = new ConnectorAsServer4Gate();
			startAsServer();
		}
		
		addShutdownHook();
	}

	@Override
	public boolean isAllMatch() {
		return isAllMatch;
	}

	private void startAsClient(ConnectorContext context, boolean mngMode, boolean isAllMatch) {
		for (Iterator<Entry<String, NodeConfig>> itr = gateAddress.entrySet().iterator(); itr.hasNext();) {
			final NodeConfig nodeConfig = itr.next().getValue();
			final int gateNo = nodeConfig.getKeyNo();
			final int pid = nodeConfig.getPid();
			final int logLevel = nodeConfig.getLogLevel();
			final String gateAddress = nodeConfig.getAddress();
			final int gatePort = nodeConfig.getPort();
			final int heartbeat = nodeConfig.getHeartbeat();
			// 线程方式启动与终端对接的服务端
			new Thread(new Runnable() {
				public void run() {
					LogWorker.logForce(String.format("%sClient模式开启，网关地址为：%s，网关端口号为：%s", ConnectorContext.CONNECTOR_NAME, gateAddress, gatePort));
					ConnectorAsClient2Gate connectorAsClient2Gate = new ConnectorAsClient2Gate(
					new ConnectionWacthDog(gateNo, pid, gateAddress, gatePort, true, new ConnectionWacthDog.Relogin() {
						@Override
						public ByteBuf getLoginBuf(Integer gateNo, Integer pid, String localAddress) throws Exception {
							return messageHandleFactory.getLoginBuf(gateNo, pid, localAddress, mngMode, isAllMatch);
						}

						@Override
						public void onConnectSuccess(String ip, Integer port, Integer pid) {
							if (isAllMatch) {
								LogWorker.logForce(String.format("%s连接网关 %s|%s:%s OK", "*", "*", ConnectorContext.CONNECTOR_NAME, ip, gatePort));
							}
							else {
								LogWorker.logForce(String.format("%s连接网关 %d|%s:%s OK", ConnectorContext.CONNECTOR_NAME, pid, ip, gatePort));
							}
							messageHandleFactory.onConnectSuccess(ip, port, pid);
						}

						@Override
						public void onConnectFailed(String ip, Integer port, Integer pid) {
							if (isAllMatch) {
								LogWorker.logForce(String.format("%s连接等待网关 %s|%s:%s Failed", "*", "*", ConnectorContext.CONNECTOR_NAME, ip, gatePort));
							}
							else {
								LogWorker.logForce(String.format("%s连接等待网关 %d|%s:%s Failed", ConnectorContext.CONNECTOR_NAME, pid, ip, gatePort));
							}
							messageHandleFactory.onConnectFailed(ip, port, pid);
						}
					}) {
						@Override
						public ChannelHandler[] getChannelHandlers() {
							return messageHandleFactory.getChannelHandlers(this, pid, logLevel, isAllMatch);
						}
						@Override
						protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
							ctx.fireChannelRead(msg);
						}
					},
					gateAddress, gatePort);
					connectorAsClient2Gate.config(heartbeat);
					ConnectorAsClient2Gate.put(pid, connectorAsClient2Gate);
				}
			}, "startAsClientThread_ip_" + gateAddress + ":" + gatePort).start();
		}
	}

	private void startAsServer(){
		new Thread(new Runnable() {
			public void run() {
				LogWorker.logForce(String.format("%s服务模式模式开启，服务端口号为：%s，心跳周期为：%sH", ConnectorContext.CONNECTOR_NAME, connectorServerPort, 1));
				connectorAsServer4Gate.bindAddress(connectorAsServer4Gate.config(context, connectorServerPid, connectorServerAddress, connectorServerPort, timeout, logLevel));
			}
		}, "startAsServerThread").start();
	}

	private void parseConfig(String baseDir) {
        final String workDir;
        final String CONFIG_FILE_NAME = "iot-connector.conf";
		if (baseDir == null) {
        	String currentDir = System.getProperty("user.dir");
        	if (!new File(currentDir).isDirectory()) {
        		throw new RuntimeException(currentDir + " 当前目录没有读权限");
        	}
            if (new File(currentDir + "/src/main/resources").isDirectory()){
            	workDir = currentDir  + File.separatorChar + "src" +  File.separatorChar + "main" + File.separatorChar + "resources" + File.separatorChar;
            }
            else if (new File(currentDir + "/config").isDirectory()){
            	workDir = currentDir + File.separatorChar + "config" + File.separatorChar;
            }
            else {
            	//默认取工作目录
            	workDir = currentDir + File.separatorChar;
            }
        }
        else {
        	if (baseDir.endsWith(CONFIG_FILE_NAME)) {
        		baseDir.substring(0, baseDir.length() - CONFIG_FILE_NAME.length());
        	}
        	if (!new File(baseDir).isDirectory()) {
        		throw new RuntimeException(baseDir + " 指定目录没有读权限");
        	}
        	workDir = baseDir;
        }
        
    	if (!new File(workDir).isDirectory()) {
    		throw new RuntimeException(workDir + " 目录没有读权限");
    	}
    	
		try {
			Map<String, Object> config = ConfigTool.readConfigParams(workDir + CONFIG_FILE_NAME);
			@SuppressWarnings("unchecked")
			Map<String, String> common = (Map<String, String>) config.remove("common");
			parseCommonConfig(common);
			parseProtocol(config);
		} catch (Exception e) {
			e.printStackTrace();
			System.err.println("配置文件加载失败");
			System.exit(-1);
		}
	}

	private void parseCommonConfig(Map<String, String> common) {
		workerThreads = Integer.parseInt(common.get("workerThreads"));
		LogWorker.logable = !ProtocalStrategyCache.parseBoolean(common.get("nolog"));
		
		this.connectMode = common.get("connectMode");
		this.devMode = ProtocalStrategyCache.parseBoolean(common.get("devMode"));
		this.mngMode = ProtocalStrategyCache.parseBoolean(common.get("mngMode"));
		this.isAllMatch = ProtocalStrategyCache.parseBoolean(common.get("matchAllPid"));
	}

	@SuppressWarnings("unchecked")
	public void parseProtocol(Map<String, Object> config) {
		try {
			System.out.println("配置项个数：" + config.size());
			for (String key : config.keySet()) {
				if (key.startsWith("protocol")) {
					Map<String, String> paramMap = (Map<String, String>) config.get(key);
					if (paramMap.containsKey("enabled") && !ProtocalStrategyCache.parseBoolean(paramMap.get("enabled"))) {
						continue;
					}

					ProtocolConfig protocolConfig = ProtocalStrategyCache.parseProtocolConfig(paramMap);
					ProtocalStrategyCache.protocalStrategyCache.put(protocolConfig.getPid(), protocolConfig);
				}
				else if (key.startsWith("gate")) {
					Map<String, String> paramMap = (Map<String, String>) config.get(key);
					if (paramMap.containsKey("enabled") && !ProtocalStrategyCache.parseBoolean(paramMap.get("enabled"))) {
						continue;
					}
					boolean workAsClient = Boolean.parseBoolean(paramMap.get("workAsClient"));
					int keyNo = Integer.parseInt(paramMap.get("keyNo"));
					int pid = Integer.parseInt(paramMap.get("pid"));
					String address = paramMap.get("address");
					int port = Integer.parseInt(paramMap.get("port"));
					int heartbeat = paramMap.containsKey("heartbeat") ? 2 : Integer.parseInt(paramMap.get("heartbeat"));
					if (workAsClient) {
						connectAsClient = true;
						HostRuntimeCache.gateNo = keyNo;
						LogWorker.logForce(String.format("网关编号为：%s", HostRuntimeCache.gateNo));
						timeout = heartbeat;
						connectorServerPid = pid;
						connectorServerAddress = address;
						connectorServerPort = port;
					}
					else {
						connectAsClient = false;
					}
					
					NodeConfig nodeConfig = new NodeConfig(workAsClient, keyNo, pid, address, port);
					nodeConfig.setHeartbeat(heartbeat);
					logLevel = paramMap.containsKey("logLevel") ? Integer.parseInt(paramMap.get("logLevel")) : 0;
					nodeConfig.setLogLevel(logLevel);
					gateAddress.put(address + ":" + port + ":" + pid, nodeConfig);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
			System.err.println("配置文件加载失败");
			System.exit(-1);
		}
	}

	/**
	 * JVM的关闭钩子--JVM正常关闭才会执行
	 */
	private void addShutdownHook(){
		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
			public void run() {
				shutdown();
			}
		}));
	}

	@Override
	public void shutdown() {
		//清空缓存信息
		LogWorker.logForce(ConnectorContext.CONNECTOR_NAME + "正常关闭前执行  清空所有缓存信息...............................");
		LogWorker.logForce("Gate2ConnectorWorker.getInstance().stopWorker()");
		Gate2ConnectorWorker.getInstance().stopWorker();

		for(Iterator<Entry<Integer, ConnectorAsClient2Gate>> itr = ConnectorAsClient2Gate.entrySet().iterator(); itr.hasNext(); ) {
			LogWorker.logForce("ConnectorAsClient2Gate = itr.next().getValue().close()");
			itr.next().getValue().close();
		}
		
		if (connectorAsServer4Gate != null) {
			LogWorker.logForce("connectorAsServer2Gate.close()");
			connectorAsServer4Gate.close();
		}

		LogWorker.logForce("Connector2GateWorker.getInstance().stopWorker()");
		Connector2GateWorker.getInstance().stopWorker();
		
		int times = 10;
		while (!Gate2ConnectorWorker.getInstance().isShutdown() || !Connector2GateWorker.getInstance().isShutdown()) {
			try {
				Thread.sleep(1000L);
			} catch (InterruptedException e) {
			}
			times--;
			if (times < 0) {
				break;
			}
		}

		messageHandleFactory.stopMQListener();
		LogWorker.logForce(String.format("%s正常关闭前执行  清空所有缓存信息 times%s...............................", ConnectorContext.CONNECTOR_NAME, times));
	}
	
	/**
	 * @return the hostApiAddress
	 */
	@Override
	public String getConnectorApiAddress() {
		return connectorApiAddress;
	}
}